package spring.cloud.tasks.common.curator;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import spring.cloud.tasks.common.CuratorUtils;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class CuratorImpl implements Curator {
    public static final String EMPTY_STRING = "";
    private CuratorInfo curatorInfo;
    private final CuratorFramework curatorFramework;
    private long sessionId;

    public CuratorImpl(CuratorInfo curatorInfo, CuratorFramework curatorFramework) throws RuntimeException {
        this.curatorInfo = curatorInfo;
        this.curatorFramework = curatorFramework;
        try {
            this.sessionId = curatorFramework.getZookeeperClient().getZooKeeper().getSessionId();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    @Override
    public CuratorInfo getCuratorInfo() {
        return curatorInfo;
    }

    @Override
    public CuratorFramework getCuratorFramework() {
        return curatorFramework;
    }

    @Override
    public long getSessionId() {
        return sessionId;
    }

    @Override
    public boolean isConnected() {
        return curatorFramework != null && curatorFramework.getZookeeperClient().isConnected();
    }

    @Override
    public void close() {
        CloseableUtils.closeQuietly(curatorFramework);
    }

    //
    @Override
    public void addConnectionStateListener(final ConnectionStateListener connectionStateListener) {
        curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
    }

    @Override
    public void removeConnectionStateListener(final ConnectionStateListener connectionStateListener) {
        curatorFramework.getConnectionStateListenable().removeListener(connectionStateListener);
    }

    /**
     * 共用一个连接会话,开启事务后进行事务操作
     */
    @Override
    public TransactionalCurator getTransactionalCurator() {
        return new TransactionalCuratorImpl(curatorFramework);
    }

    //
    @Override
    public boolean checkExists(final String path) {
        try {
            return null != curatorFramework.checkExists().forPath(path);
        } catch (Exception e) {
            log.error(e.getMessage());
            throw new RuntimeException(e);
        }
    }

    @Override
    public Stat getStat(String path) {
        try {
            Stat stat = curatorFramework.checkExists().forPath(path);
            return stat;
        } catch (final Exception e) {
            log.error(e.getMessage());
            throw new RuntimeException(e);
        }
    }

    @Override
    public Long getCtime(String path) {
        try {
            Stat stat = curatorFramework.checkExists().forPath(path);
            if (stat == null) {
                return null;
            } else {
                return stat.getCtime();
            }
        } catch (final Exception ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public Long getMtime(String path) {
        try {
            Stat stat = curatorFramework.checkExists().forPath(path);
            if (stat == null) {
                return null;
            } else {
                return stat.getMtime();
            }
        } catch (final Exception ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public String getData(final String path) {
        try {
            if (!checkExists(path)) {
                return null;
            } else {
                byte[] bytes = curatorFramework.getData().forPath(path);
                if (bytes == null) {//瞬间节点被删除,需要尽量让ZK操作避免瞬时被删除。容易引起错误处理。如何出现了这种情况，尽量在分布式锁的情况进行删除。
                    return null;
                }
                return new String(bytes, StandardCharsets.UTF_8);
            }
        } catch (final KeeperException.NoNodeException e) {//瞬间节点被删除
            return null;
        } catch (final Exception ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public List<String> getChildren(final String path) {
        try {
            return curatorFramework.getChildren().forPath(path);
        } catch (final KeeperException.NoNodeException e) {//瞬间节点被删除
            return null;
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void ephemeralCreate(final String path, final String value) {
        try {
            /**
             * 如果存在则先删除后再创建节点
             */
            if (checkExists(path)) {
                curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
            }
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, value.getBytes(StandardCharsets.UTF_8));
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void create(final String path) {
        create(path, EMPTY_STRING);//***需要注意的是默认值为""***
    }

    @Override
    public void create(final String path, @NonNull String value) {//createNodeAndSetValue  add by huihui
        try {
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, value.getBytes(StandardCharsets.UTF_8));
        } catch (final KeeperException.NodeExistsException e) {
        } catch (final Exception ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public void update(final String path, final String value) {
        try {
            curatorFramework.inTransaction().check().forPath(path).and().setData().forPath(path, value.getBytes(StandardCharsets.UTF_8)).and().commit();
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void merge(final String path, @NonNull final String value) {
        if (this.checkExists(path)) {
            update(path, value);
        } else {
            this.create(path, value);
        }
    }

    @Override
    public void createIfAbsent(final String path, @NonNull final String value) {
        if (this.checkExists(path)) {
            //如果已经存在了则不进行更新。这种是可重复执行才调用这个方法
        } else {
            this.create(path, value);
        }
    }

    @Override
    public void delete(final String path) {
        try {
            curatorFramework.delete().guaranteed().forPath(path);
        } catch (KeeperException.NotEmptyException e) {
            log.warn("当前节点可能是目录:" + path + ",进行递归删除中", e);
            try {
                CuratorUtils.deleteRecursive(curatorFramework, path);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        } catch (KeeperException.NoNodeException e) {
            log.warn("当前节点不存在:" + path + ",该操作直接忽略", e);
        } catch (final Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }
}
